/*  
 * @(#) KafkaFetchTest.java Create on 2015年1月28日 上午10:56:32   
 *   
 * Copyright 2015 jstx.
 */

package kafka.examples;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.common.ErrorMapping;
import kafka.common.OffsetMetadataAndError;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetCommitRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.Message;
import kafka.message.MessageAndOffset;
/**
 * @KafkaFetchTest.java
 * @created at 2015年1月28日 上午10:56:32 by zhanghongliang@hiveview.com
 *
 * @desc
 *
 * @author zhanghongliang@hiveview.com
 * @version $Revision$
 * @update: $Date$
 */
public class KafkaFetchTest {
	public static String getMessage(Message message) {
		ByteBuffer buffer = message.payload();
		byte[] bytes = new byte[buffer.remaining()];
		buffer.get(bytes);
		return new String(bytes);
	}

	public static void main(String[] args) {
		String clientName = "test";
		String a_topic = "topic3";
		SimpleConsumer consumer = new SimpleConsumer("hdp5", 9092, 10000,
				1024000,clientName);
		long offset = 0L;
		int partition = 0;
		while (true) {
			//			FetchRequest fetchRequest = new FetchRequest("test", 0, 0L, 100000);
			FetchRequest req = new FetchRequestBuilder().clientId(clientName)
					.addFetch(a_topic, partition, offset, 300).build();
			FetchResponse messages = consumer.fetch(req);
			for (MessageAndOffset messageAndOffset : messages.messageSet(a_topic, 0)) {
				System.out.println(getMessage(messageAndOffset.message()));
				offset = messageAndOffset.offset()+1;
			}
			
			/**
			 * 单个kafka 可以使用批量拉取，集群中会有问题
			 */
			String groupId = clientName;
			Map<TopicAndPartition, OffsetMetadataAndError> requestInfo = new HashMap<TopicAndPartition, OffsetMetadataAndError>();
			OffsetMetadataAndError value = new OffsetMetadataAndError(offset, OffsetMetadataAndError.NoMetadata(), ErrorMapping.NoError());
			TopicAndPartition key = new TopicAndPartition(a_topic, partition);
			requestInfo.put(key , value);
			short versionId = kafka.api.OffsetCommitRequest.CurrentVersion();
			int correlationId = 0;
			String clientId = clientName;
			OffsetCommitRequest offsetCommitRequest = new OffsetCommitRequest(groupId , requestInfo, versionId, correlationId, clientId );
			consumer.commitOffsets(offsetCommitRequest );
			System.out.println("-0------------------------------");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}
